package mqtt5;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttClientInterface;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.springframework.stereotype.Component;

/**
 * @Auther: 陈立
 * @Date: 2018/11/21 14:36
 * @Description: 发布消息的回调类
 * <p>
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 * <p>
 * public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 * <p>
 * public void connectionLost(Throwable cause)在断开连接时调用。
 * <p>
 * public void deliveryComplete(MqttDeliveryToken token))
 * 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
 * 由 MqttClient.connect 激活此回调。
 */
@Component
@Slf4j
@Data
public class PubCallback implements MqttCallback
{
    public String clientName;
//    @Autowired
//    @Qualifier("testService")
//    private MqttCallbackService mqttCallbackService; //= ApplicationEngine.getBean(MqttCallbackServiceImpl.class);

    @Override
    public void disconnected(MqttDisconnectResponse mqttDisconnectResponse) {
        log.info("MQTT 连接断开。");
        System.out.println("MQTT connection lost!!!!!!!!!!!!");

        // 连接丢失后，一般在这里面进行重连
        boolean flag = true;
        Mqtt5Client client = MQTTCLIENTS.CLIENTSMAP.get(clientName);
        while (flag)
        {
            log.info("连接断开，可以做重连");
            try
            {
                Thread.sleep(client.getMqttConfig().getReconnectMillisecond());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            log.info("判断是否连接 = " + client.isConnected());
            client.reconnect();
            if (client.isConnected())
            {
                flag = false;
            }
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception
    {
        // subscribe后得到的消息会执行到这里面
        String data = new String(mqttMessage.getPayload());
        if (data == null || "".equals(data))
        {
            log.info("接收到空白消息,不予处理");
        }
        try
        {
            log.info("data = " + data);
            System.out.println("data = " + data);
//            mqttCallbackService.task(data);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    @Override
    public void deliveryComplete(IMqttToken iMqttToken) {
        MqttClientInterface client = iMqttToken.getClient();
        String clientId = client.getClientId();
        String url = client.getServerURI();
        String[] topics = iMqttToken.getTopics();
        log.info("发布消息的ID=" + clientId + ";地址=" + url);
        StringBuffer str = new StringBuffer("");
        if (topics != null)
        {
            for (int i = 0; i < topics.length; i++)
            {
                str.append(topics[i] + ";");
            }
        }
        log.info("deliveryComplete---------" + iMqttToken.isComplete() + "----主题：" + str.toString());
    }

    @Override
    public void connectComplete(boolean b, String s) {
        System.out.println(" MQTT5 connectComplete ");
    }

    @Override
    public void authPacketArrived(int i, MqttProperties mqttProperties) {
        System.out.println("PubCallback.authPacketArrived");
        System.out.println("i = " + i);
        System.out.println("mqttProperties = " + mqttProperties);
    }

    @Override
    public void mqttErrorOccurred(MqttException e) {
        System.out.println("PubCallback.mqttErrorOccurred");
        System.out.println("e.getMessage() = " + e.getMessage());
        e.printStackTrace();
    }

}
